package com.example.service.pulsar;

import com.example.ececption.PulsarBusinessException;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @author Administrator
 */
@Slf4j
public class PulsarService {

    private Map<String, PulsarClient> clients = new HashMap<>(8);

    public void setClients(Map<String, PulsarClient> clients) {
        this.clients = clients;
    }

    /**
     * 同步发送消息
     *
     * @param msg 消息
     * @return 消息ID
     */
    public MessageId send(String id, String topic, String msg) throws ExecutionException, InterruptedException, PulsarClientException {
        try {
            MessageId messageId = this.sendAsync(id, topic, msg).get();
            log.info("[Pulsar] Producer同步发送消息成功，msg is {}", msg);
            return messageId;
        } catch (InterruptedException | ExecutionException | PulsarClientException e) {
            log.error("[Pulsar] Producer同步发送消息失败，msg is {}", msg);
            throw e;
        }
    }

    /**
     * 异步发送消息
     *
     * @param msg 消息
     * @return CompletableFuture
     */
    public CompletableFuture<MessageId> sendAsync(String id, String topic, String msg) throws PulsarClientException {
        Producer<String> stringProducer = null;
        try {
            if (clients.containsKey(id)) {
                PulsarClient client = clients.get(id);
                if (Objects.isNull(client)) {
                    log.error("[Pulsar] 数据源对应PulsarClient不存在，sourceName id {}", id);
                    throw new PulsarBusinessException("[Pulsar] 数据源对应PulsarClient不存在！");
                }
                stringProducer = client.newProducer(Schema.STRING).topic(topic).create();
            }
            return stringProducer.sendAsync(msg);
        } catch (Exception e) {
            log.error("Pulsar sendAsync msg {} failed!", msg);
            throw e;
        }
    }

}
